package com.beiding.jsonanalyzer.flow;

import com.beiding.jsonanalyzer.analyze.KeyMode;
import com.beiding.jsonanalyzer.analyze.KeyUtils;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * todo 1 记录父节点,父节点又记录每个子节点. 由此可从任意节点遍历全部节点
 * todo 2 流程终止反馈
 * todo 3 背压控制
 *
 * todo 去掉分支上的观察者,仅保留flow上的观察者
 *
 * @param <T>
 */
public class Branch<T> {


    /**
     * 计算源 可对后继节点复用
     */
    private ComputeResource computeResource;

    /**
     * 任务槽 唯一绑定的
     */
    private TaskSlot taskSlot;

    /**
     * 分支依据
     */
    private Object refer;

    /**
     * 分支key
     */
    private Object referKey;

    private Map<Object, Object> referMap = new HashMap<>();

    /**
     * 附加物
     */
    private  Map<Class, Object> additionMap = new HashMap<>();

    private String name;

    /**
     * 父级分支
     */
    private Branch<T> parent;

    /**
     * 流程节点工厂
     */
    private ValueNodeFactory<T> valueNodeFactory;

    /**
     * 子分支
     */
    private List<Branch<T>> children = Collections.synchronizedList(new ArrayList<>());

    /**
     * 第一个节点
     */
    private BranchNode<T> firstNode;

    /**
     * 分支依据链
     */
    private List<Object> branchReferLink;

    Branch() {

    }

    void setName(String name) {
        this.name = name;
    }

    public String getName() {
        return name;
    }

    void setReferKeyValue(Object referKey, Object value) {
        this.referKey=referKey;
        this.refer = value;
        this.referMap.put(KeyUtils.handleKey(referKey, KeyMode.TEXT_TO_KEY), value);

    }

    public Object getReferKey() {
        return referKey;
    }

    public BranchNode<T> getFirstNode() {
        return firstNode;
    }

    public List<Branch<T>> getChildren() {
        return Collections.unmodifiableList(children);
    }

    public Branch<T> getParent() {
        return parent;
    }

    public ComputeResource getComputeResource() {
        return computeResource;
    }


    public TaskSlot getTaskSlot() {
        return taskSlot;
    }


    public Object getRefer() {
        return refer;
    }

    public Object getReferByKey(Object key) {
        return referMap.get(KeyUtils.handleKey(key, KeyMode.TEXT_TO_KEY));
    }

    public List<Object> getBranchReferLink() {
        return branchReferLink;
    }


    void setTaskSlot(TaskSlot taskSlot) {
        this.taskSlot = taskSlot;
    }

    void setComputeResource(ComputeResource computeResource) {
        this.computeResource = computeResource;
    }

    void setParent(Branch<T> parent) {
        this.parent = parent;
        if (parent != null) {
            this.referMap.putAll(parent.referMap);
        }
    }

    synchronized void setChildren(List<Branch<T>> children) {
        this.children = children;
    }

    /**
     * 添加一个子分支
     *
     * @param child
     */
    synchronized void addChild(Branch<T> child) {
        children.add(child);
    }

    void setFirstNode(BranchNode<T> firstNode) {
        this.firstNode = firstNode;
    }

    void setBranchReferLink(List<Object> branchReferLink) {
        this.branchReferLink = branchReferLink;
    }

    void setValueNodeFactory(ValueNodeFactory<T> valueNodeFactory) {
        this.valueNodeFactory = valueNodeFactory;
    }

    public ValueNodeFactory<T> getValueNodeFactory() {
        return valueNodeFactory;
    }

    /**
     * 添加观察者
     *
     * @param type
     * @param eventListener
     * @param <E>
     */
    public <E> void addListener(Class<E> type, EventListener<E> eventListener) {

        synchronized (getEventListenerMap()) {
            Map<Class<E>, List<EventListener<E>>> eventListenerMap = getEventListenerMap();

            List<EventListener<E>> eventListeners = eventListenerMap.computeIfAbsent(type, k -> Collections.synchronizedList(new ArrayList<>()));
            eventListeners.add(eventListener);
        }

    }

    /**
     * 移除观察者
     *
     * @param type
     * @param eventListener
     * @param <E>
     */
    public <E> void removeListener(Class<E> type, EventListener<E> eventListener) {
        synchronized (getEventListenerMap()) {
            Map<Class<E>, List<EventListener<E>>> eventListenerMap = getEventListenerMap();
            List<EventListener<E>> eventListeners = eventListenerMap.get(type);
            if (eventListeners != null) {
                eventListeners.remove(eventListener);
                if (eventListeners.size() == 0) {
                    eventListenerMap.remove(type);
                }
            }
        }
    }

    /**
     * 获取某类型的所有观察者
     *
     * @param type
     * @param <E>
     * @return
     */
    public <E> List<EventListener<E>> getListeners(Class<E> type) {
        Map<Class<E>, List<EventListener<E>>> eventListenerMap = getEventListenerMap();
        List<EventListener<E>> eventListeners = eventListenerMap.get(type);
        if (eventListeners == null) {
            return Collections.emptyList();
        }
        return Collections.unmodifiableList(eventListeners);
    }

    /**
     * 提交事件
     *
     * @param event
     */
    public void emit(Object event) {

        executeEventSelf(event);

        //反向广播给父节点

        if (this.parent != null) {
            this.parent.emitFromChild(this, event);
        }

    }

    /**
     * 接收到从子节点发出的事件
     *
     * @param from
     * @param event
     */
    private void emitFromChild(Branch<T> from, Object event) {

        //先自行执行
        this.executeEventSelf(event);

        //广播给出了来源分支外的其他分支
        synchronized (this) {
            if (this.children != null && this.children.size() > 0) {
                for (Branch<T> branch : this.children) {
                    if (branch != from) {
                        branch.emitFromParent(event);
                    }
                }
            }
        }

        //再反向广播给父节点

        if (this.parent != null) {
            this.parent.emitFromChild(this, event);
        }

    }

    /**
     * 接收到来自父节点的事件
     *
     * @param event
     */
    private void emitFromParent(Object event) {
        executeEventSelf(event);
        //再广播给下级

        synchronized (this) {

            if (this.children != null && this.children.size() > 0) {
                for (Branch<T> branch : this.children) {
                    branch.emitFromParent(event);
                }
            }
        }

    }

    /**
     * 事件自己执行
     *
     * @param event
     */
    private void executeEventSelf(Object event) {
        synchronized (getEventListenerMap()) {

            //不能为空
            Class type = event.getClass();
            Map<Class<Object>, List<EventListener<Object>>> eventListenerMap = getEventListenerMap();
            for (Map.Entry<Class<Object>, List<EventListener<Object>>> entry : eventListenerMap.entrySet()) {
                Class tp = entry.getKey();

                //执行动作
                if (tp.isAssignableFrom(type)) {
                    List<EventListener<Object>> value = entry.getValue();
                    for (EventListener<Object> listener : value) {
                        try {
                            listener.on(event);
                        } catch (Exception e) {
                            //只打印异常,不会终止流程
                            e.printStackTrace();
                        }
                    }
                }
            }

        }
    }

    private Map eventListenerMap = new ConcurrentHashMap();

    private <E> Map<Class<E>, List<EventListener<E>>> getEventListenerMap() {
        return eventListenerMap;
    }

    /**
     * 当前最大节点数
     */
    private int breakPressureSize = 0;

    /**
     * 从子分支推算出来的背压数
     * 自身设置的背压数优先级更高
     */
    private int sumChildrenBreakPressureSize = 0;

    void breakPressure(int size) {

        //设置锁,如果父分支存在则使用父分支,否则使用自身
        //防止反推给父节点的时候,其他子分支被修改
        Object lock = parent != null ? parent : this;

        synchronized (lock) {

            this.breakPressureSize = size;

            //从首节点上施加控制
            DataFlow dataFlow = this.firstNode.getDataFlow();

            //先控制当前节点
            dataFlow.maxEventCount(size);

            //反推给父分支

            //父节点处理原则: 如果所有的子分支都被背压,则父节点背压值为所有子节点背压值的和;如果有一个子分支未背压,则父节点无背压
            if (parent != null) {
                parent.breakPressureFromChild();
            }
        }


    }


    private void breakPressureFromChild() {

        //防止父分支处理时子分支修改
        //最终整个反推会传递到根分支,使整个分支树都加锁
        Object lock = parent != null ? parent : this;
        synchronized (lock) {
            synchronized (children) {
                int sum = 0;

                for (Branch<T> child : children) {
                    //子节点自身的背压数字为准. 如果子节点没有设置背压,则取用推算出来的背压数字
                    int n = child.breakPressureSize > 0 ? child.breakPressureSize : child.sumChildrenBreakPressureSize > 0 ? child.sumChildrenBreakPressureSize : 0;
                    sum += n;
                }

                sumChildrenBreakPressureSize = sum;

                //如果自身未设置背压,则取用推算的背压
                if (breakPressureSize <= 0) {
                    DataFlow dataFlow = this.firstNode.getDataFlow();
                    dataFlow.maxEventCount(sumChildrenBreakPressureSize);
                }

                if (parent != null) {
                    parent.breakPressureFromChild();
                }
            }
        }

    }

    int getBreakPressureSize() {
        return breakPressureSize;
    }

    /**
     * 添加一个附加物
     * @param w
     * @param <W>
     * @return
     */
    public <W> W addition(W w) {
        return (W) additionMap.put(w.getClass(), w);
    }

    public <W> W removeAddition(Class<W> type) {
        return (W) additionMap.remove(type);
    }

    public <W> W getAddition(Class<W> type) {
        return (W) additionMap.get(type);
    }

}
